#if !defined(_WIN32)

#include	<Network.h>
#include	<Logger.h>

#include	<algorithm>
#include	<cstdlib>
#include	<cstring>
#include	<map>
#include	<thread>
#include	<vector>

#include	<arpa/inet.h>
#include	<fcntl.h>
#include	<netinet/in.h>
#include	<sys/epoll.h>
#include	<sys/socket.h>
#include	<sys/types.h>
#include	<unistd.h>

#define		SOCKET_BUFSIZE	2097152

using namespace std;

class Socket {
public:
	Socket(IConnector * pOwner);
	virtual ~Socket();

	int		Connect(const string & sIP, int nPort);
	bool	IsConnected() { return _nSocket >= 0; }
	void	Close(ENet::Close emCode);
	bool	Send(const char * pData, size_t nSize);
	void	Breath();

private:
	IConnector *	_pOwner;
	char *			_pReceived;
	int				_nSocket;
	int				_nIO;
};

Socket::Socket(IConnector * pOwner)
	: _pOwner(pOwner)
	, _pReceived(new char[SOCKET_BUFSIZE])
	, _nSocket(-1)
	, _nIO(0) {}

Socket::~Socket() {
	Close(ENet::Local);
	delete[] _pReceived;
}

int Socket::Connect(const string & sIP, int nPort) {
	if (_nSocket >= 0) return ENet::Running;
	if ((_nSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) return errno;

	int nFlags = fcntl(_nSocket, F_GETFL, 0);
	if (fcntl(_nSocket, F_SETFL, O_NONBLOCK | nFlags) < 0) {
		close(_nSocket);
		_nSocket = -1;
		return errno;
	}

	struct sockaddr_in iAddr;
	memset(&iAddr, 0, sizeof(iAddr));

	iAddr.sin_family	= AF_INET;
	iAddr.sin_port		= htons(nPort);

	if (0 >= inet_pton(AF_INET, sIP.c_str(), &iAddr.sin_addr)) {
		close(_nSocket);
		_nSocket = -1;
		return ENet::BadParam;
	}

	if (connect(_nSocket, (sockaddr *)&iAddr, sizeof(sockaddr)) < 0) {
		if (errno != EINPROGRESS) {
			close(_nSocket);
			_nSocket = -1;
			return errno;
		}

		fd_set iSet;
		struct timeval iWait;

		iWait.tv_sec = 3;
		iWait.tv_usec = 0;

		FD_ZERO(&iSet);
		FD_SET(_nSocket, &iSet);

		if (select(_nSocket + 1, 0, &iSet, 0, &iWait) <= 0) {
			close(_nSocket);
			_nSocket = -1;
			return ENet::Timeout;
		}

		int nErr;
		socklen_t nLen;
		if (getsockopt(_nSocket, SOL_SOCKET, SO_ERROR, &nErr, &nLen) < 0 || nErr != 0) {
			close(_nSocket);
			_nSocket = -1;
			return ENet::Closed;
		}
	}

	if ((_nIO = epoll_create(1)) < 0) {
		close(_nSocket);
		_nSocket = -1;
		return ENet::Epoll;
	}

	struct epoll_event iEv;
	iEv.events = EPOLLIN | EPOLLET;
	iEv.data.fd = _nSocket;
	if (epoll_ctl(_nIO, EPOLL_CTL_ADD, _nSocket, &iEv) < 0) {
		close(_nIO);
		close(_nSocket);
		_nSocket = -1;
		return ENet::Epoll;
	}

	_pOwner->OnConnected();
	return 0;
}

void Socket::Close(ENet::Close emCode) {
	if (_nSocket < 0) return;
	close(_nIO);
	close(_nSocket);
	_nSocket = -1;
	_pOwner->OnClose(emCode);
}

bool Socket::Send(const char * pData, size_t nSize) {
	if (_nSocket < 0) return false;

	char *	pSend	= (char *)pData;
	int		nSend	= 0;
	int		nLeft	= (int)nSize;

	while (true) {
		nSend = (int)send(_nSocket, pSend, nLeft, MSG_DONTWAIT);
		if (nSend < 0) {
			if (errno == EAGAIN) {
				usleep(1000);
			} else {
				return false;
			}
		} else if (nSend < nLeft) {
			nLeft -= nSend;
			pSend += nSend;
		} else if (nSend == nLeft) {
			return true;
		} else {
			return nLeft == 0;
		}
	}
}

void Socket::Breath() {
	if (_nSocket < 0) return;

	static epoll_event pEvents[4] = { 0 };
	int nCount = epoll_wait(_nIO, pEvents, 4, 0);
	if (nCount <= 0) return;

	memset(_pReceived, 0, SOCKET_BUFSIZE);
	int nReaded = 0;

	while (true) {
		int nRecv = (int)recv(_nSocket, _pReceived + nReaded, SOCKET_BUFSIZE - nReaded, MSG_DONTWAIT);
		if (nRecv < 0) {
			if (errno == EAGAIN) {
				break;
			} else {
				Close(ENet::BadData);
				break;
			}
		} else if (nRecv == 0) {
			Close(ENet::Remote);
			break;
		} else {
			nReaded += nRecv;
		}
	}

	if (nReaded > 0) _pOwner->OnReceive(_pReceived, nReaded);
}

class ServerSocket {
	typedef map<uint64_t, Connection *> ConnectionMap;

public:
	ServerSocket(IService * pOwner);
	virtual ~ServerSocket();

	int		Listen(const string & sIP, int nPort);
	void	Broadcast(const char * pData, size_t nSize);
	void	Close(Connection * pConn, ENet::Close emCode);
	void	Shutdown();
	void	Breath();

	Connection *	Find(uint64_t nConnId);

private:
	IService *			_pOwner;
	char *				_pReceived;
	int					_nSocket;
	ConnectionMap		_mConns;
	ConnectionMap		_mSocket2Conns;
	int					_nIO;
};

ServerSocket::ServerSocket(IService * pOwner)
	: _pOwner(pOwner)
	, _pReceived(new char[SOCKET_BUFSIZE])
	, _nSocket(-1)
	, _mConns()
	, _mSocket2Conns()
	, _nIO(0) {}

ServerSocket::~ServerSocket() {
	Shutdown();
	delete[] _pReceived;
}

int ServerSocket::Listen(const string & sIP, int nPort) {
	if (_nSocket >= 0) return ENet::Running;
	if ((_nSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) return errno;

	int nReuse = 1;
	setsockopt(_nSocket, SOL_SOCKET, SO_REUSEADDR, &nReuse, sizeof(nReuse));

	int nFlags = fcntl(_nSocket, F_GETFL, 0);
	if (fcntl(_nSocket, F_SETFL, O_NONBLOCK | nFlags) < 0) {
		close(_nSocket);
		_nSocket = -1;
		return errno;
	}

	struct sockaddr_in iAddr;
	memset(&iAddr, 0, sizeof(iAddr));

	iAddr.sin_family = AF_INET;
	iAddr.sin_port = htons(nPort);

	if (0 >= inet_pton(AF_INET, sIP.c_str(), &iAddr.sin_addr)) {
		close(_nSocket);
		_nSocket = -1;
		return ENet::BadParam;
	}

	if (::bind(_nSocket, (sockaddr *)&iAddr, sizeof(sockaddr)) < 0 || ::listen(_nSocket, 512) < 0) {
		close(_nSocket);
		_nSocket = -1;
		return errno;
	}

	if ((_nIO = epoll_create(1)) < 0) {
		close(_nSocket);
		_nSocket = -1;
		return ENet::Epoll;
	}

	struct epoll_event iEv;
	iEv.events = EPOLLIN | EPOLLET;
	iEv.data.fd = _nSocket;

	if (epoll_ctl(_nIO, EPOLL_CTL_ADD, _nSocket, &iEv) < 0) {
		close(_nIO);
		close(_nSocket);
		_nSocket = -1;
		return ENet::Epoll;
	}

	return 0;
}

void ServerSocket::Broadcast(const char * pData, size_t nSize) {
	for (auto & kv : _mConns) {
		int		nSocket	= kv.second->_nSocket;
		char *	pSend	= (char *)pData;
		int		nSend	= 0;
		int		nLeft	= (int)nSize;

		while (true) {
			nSend = (int)send(nSocket, pSend, nLeft, MSG_DONTWAIT);
			if (nSend < 0) {
				if (errno == EAGAIN) {
					usleep(1000);
				} else {
					break;
				}
			} else if (nSend < nLeft) {
				nLeft -= nSend;
				pSend += nSend;
			} else {
				break;
			}
		}
	}
}

void ServerSocket::Close(Connection * pConn, ENet::Close emCode) {
	if (!pConn) return;

	int nSocket = pConn->_nSocket;
	uint64_t nConnId = pConn->_nId;

	pConn->_nSocket = -1;
	_pOwner->OnClose(pConn, emCode);
	epoll_ctl(_nIO, EPOLL_CTL_DEL, nSocket, NULL);
	close(nSocket);
	delete pConn;

	_mConns.erase(nConnId);
	_mSocket2Conns.erase(nSocket);
}

void ServerSocket::Shutdown() {
	if (_nSocket < 0) return;

	for (auto & kv : _mConns) {
		Connection * pConn = kv.second;
		_pOwner->OnClose(pConn, ENet::Local);
		epoll_ctl(_nIO, EPOLL_CTL_DEL, pConn->_nSocket, NULL);
		close(pConn->_nSocket);
		delete pConn;
	}

	epoll_ctl(_nIO, EPOLL_CTL_DEL, _nSocket, NULL);
	_mConns.clear();
	_mSocket2Conns.clear();

	close(_nIO);
	close(_nSocket);
	_nSocket = -1;

	_pOwner->OnShutdown();
}

void ServerSocket::Breath() {
	if (_nSocket < 0) return;
	static epoll_event pEvents[1024] = { 0 };
	static uint64_t nAllocId = 0;
	static sockaddr_in iAddr = { 0 };
	static socklen_t nSizeOfAddr = sizeof(iAddr);
	static char pAddr[128] = { 0 };

	int nCount = epoll_wait(_nIO, pEvents, 1024, 0);
	if (nCount <= 0) return;

	for (int i = 0; i < nCount; ++i) {
		if (!(pEvents[i].events & EPOLLIN)) continue;

		if (pEvents[i].data.fd == _nSocket) {
			while (true) {
				int nAccept = accept(_nSocket, (sockaddr *)&iAddr, &nSizeOfAddr);
				if (nAccept < 0) break;

				uint64_t nConnId = nAllocId + 1;
				nAllocId++;
				inet_ntop(AF_INET, &iAddr, pAddr, 128);

				struct epoll_event iEv;
				iEv.events = EPOLLIN | EPOLLET;
				iEv.data.fd = nAccept;

				if (epoll_ctl(_nIO, EPOLL_CTL_ADD, nAccept, &iEv) < 0) {
					LOG_WARN("Failed accept client [%s] while setting non-block!!!", pAddr);
					continue;
				}

				Connection * pConn = new Connection(this, nConnId, nAccept, iAddr.sin_addr.s_addr);
				_mConns[nConnId] = pConn;
				_mSocket2Conns[(uint64_t)nAccept] = pConn;
				_pOwner->OnAccept(pConn);
			}
		} else {
			int nSocket = pEvents[i].data.fd;
			int nReaded = 0;

			auto it = _mSocket2Conns.find((uint64_t)nSocket);
			if (it == _mSocket2Conns.end()) continue;

			memset(_pReceived, 0, SOCKET_BUFSIZE);

			while (true) {
				int nRecv = (int)recv(nSocket, _pReceived + nReaded, SOCKET_BUFSIZE - nReaded, MSG_DONTWAIT);
				if (nRecv > 0) {
					nReaded += nRecv;
					if (nReaded >= SOCKET_BUFSIZE) {
						_pOwner->OnReceive(it->second, _pReceived, nReaded);
						memset(_pReceived, 0, SOCKET_BUFSIZE);
						nReaded = 0;
					}
				} else if (nRecv < 0 && errno == EAGAIN) {
					if (nReaded > 0) _pOwner->OnReceive(it->second, _pReceived, nReaded);
					break;
				} else {
					if (nReaded > 0) _pOwner->OnReceive(it->second, _pReceived, nReaded);
					Close(it->second, nRecv == 0 ? ENet::Remote : ENet::BadData);
					break;
				}
			}	
		}
	}
}

Connection * ServerSocket::Find(uint64_t nConnId) {
	auto it = _mConns.find(nConnId);
	if (it == _mConns.end()) return nullptr;
	return it->second;
}

class SocketGuard {
public:
	SocketGuard(IConnector * p, const string & sHost, int nPort);
	virtual ~SocketGuard();

	void	Start();

private:
	IConnector *	_p;
	string			_sHost;
	int				_nPort;
	thread *		_pWorker;
	bool			_bRunning;
};

SocketGuard::SocketGuard(IConnector * p, const string & sHost, int nPort)
	: _p(p), _sHost(sHost), _nPort(nPort), _pWorker(nullptr), _bRunning(false) {}

SocketGuard::~SocketGuard() {
	_bRunning = false;
	if (_pWorker) {
		if (_pWorker->joinable()) _pWorker->join();
		delete _pWorker;
	}
}

void SocketGuard::Start() {
	if (_bRunning) return;

	if (_pWorker) {
		if (_pWorker->joinable()) _pWorker->join();
		delete _pWorker;
	}
	
	_bRunning = true;
	_pWorker = new thread([this]() {
		while (_bRunning) {
			if (!_p->IsConnected()) {
				int n = _p->Connect(_sHost, _nPort, false);
				if (n != ENet::Success) LOG_WARN("Try to reconnect [%s:%d] ... %d", _sHost.c_str(), _nPort, n);
			}

			this_thread::sleep_for(chrono::seconds(1));
		}
	});
}

class NetworkBreather {
public:
	NetworkBreather() {}

	static NetworkBreather & Get();

	void	Add(IConnector * p);
	void	Add(IService * p);
	void	Del(IConnector * p);
	void	Del(IService * p);

	void	Breath();

private:
	vector<IConnector *>	_vClients;
	vector<IService *>		_vServers;
};

NetworkBreather & NetworkBreather::Get() {
	static unique_ptr<NetworkBreather> pIns;
	if (!pIns) pIns.reset(new NetworkBreather);
	return *pIns;
}

void NetworkBreather::Add(IConnector * p) {
	for (auto pClient : _vClients) {
		if (pClient == p) return;
	}

	_vClients.push_back(p);
}

void NetworkBreather::Add(IService * p) {
	for (auto pServer : _vServers) {
		if (pServer == p) return;
	}

	_vServers.push_back(p);
}

void NetworkBreather::Del(IConnector * p) {
	auto it = find(_vClients.begin(), _vClients.end(), p);
	if (it != _vClients.end()) _vClients.erase(it);
}

void NetworkBreather::Del(IService * p) {
	auto it = find(_vServers.begin(), _vServers.end(), p);
	if (it != _vServers.end()) _vServers.erase(it);
}

void NetworkBreather::Breath() {
	for (auto p : _vClients) p->Breath();
	for (auto p : _vServers) p->Breath();
}

void AutoNetworkBreath() {
	NetworkBreather::Get().Breath();
}

IConnector::IConnector() : _pCtx(nullptr), _pGuard(nullptr) {
	_pCtx = new Socket(this);
	NetworkBreather::Get().Add(this);
}

IConnector::~IConnector() {
	NetworkBreather::Get().Del(this);
	Close();
	if (_pGuard) delete _pGuard;
	if (_pCtx) delete _pCtx;
}

int IConnector::Connect(const std::string & sIP, int nPort, bool bAutoReconnect /* = false */) {
	if (sIP.empty() || nPort < 0) return ENet::BadParam;

	int n = _pCtx->Connect(sIP, nPort);
	if (n == 0 && bAutoReconnect && !_pGuard) {
		_pGuard = new SocketGuard(this, sIP, nPort);
		_pGuard->Start();
	}

	return n;
}

bool IConnector::IsConnected() {
	return _pCtx->IsConnected();
}

void IConnector::Close() {
	_pCtx->Close(ENet::Local);

	if (_pGuard) {
		delete _pGuard;
		_pGuard = nullptr;
	}
}

bool IConnector::Send(const char * pData, size_t nSize) {
	if (!pData || nSize <= 0) return false;
	return _pCtx->Send(pData, nSize);
}

void IConnector::Breath() {
	_pCtx->Breath();
}

IService::IService() : _pCtx(nullptr) {
	_pCtx = new ServerSocket(this);
	NetworkBreather::Get().Add(this);
}

IService::~IService() {
	NetworkBreather::Get().Del(this);
	Shutdown();
	if (_pCtx) delete _pCtx;
}

int IService::Listen(const std::string & sIP, int nPort) {
	if (sIP.empty() || nPort < 0) return ENet::BadParam;
	return _pCtx->Listen(sIP, nPort);
}

void IService::Broadcast(const char * pData, size_t nSize) {
	if (!pData || nSize <= 0) return;
	_pCtx->Broadcast(pData, nSize);
}

void IService::Shutdown() {
	_pCtx->Shutdown();
}

void IService::Breath() {
	_pCtx->Breath();
}

Connection * IService::Find(uint64_t nConnId) {
	return _pCtx->Find(nConnId);
}

Connection::Connection(ServerSocket * pOwner, uint64_t nId, int nSocket, uint32_t nIP)
	: _pOwner(pOwner)
	, _nId(nId)
	, _nSocket(nSocket)
	, _sIP{0}
	, _pUserData(nullptr) {
	int n1 = (nIP & 0xFF);
	int n2 = (nIP >> 8) & 0xFF;
	int n3 = (nIP >> 16) & 0xFF;
	int n4 = (nIP >> 24) & 0xFF;

	snprintf(_sIP, 16, "%d.%d.%d.%d", n1, n2, n3, n4);
}

bool Connection::Send(const char * pData, size_t nSize) {
	if (_nSocket < 0) return false;

	char *	pSend	= (char *)pData;
	int		nSend	= 0;
	int		nLeft	= (int)nSize;

	while (true) {
		nSend = (int)send(_nSocket, pSend, nLeft, MSG_DONTWAIT);
		if (nSend < 0) {
			if (errno == EAGAIN) {
				usleep(1000);
			} else {
				return false;
			}
		} else if (nSend < nLeft) {
			nLeft -= nSend;
			pSend += nSend;
		} else if (nSend == nLeft) {
			return true;
		} else {
			return nLeft == 0;
		}
	}
}

void Connection::Close() {
	if (_pOwner) _pOwner->Close(this, ENet::Local);
}

#endif
